package com.imooc.ecommerce.kafka;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.imooc.ecommerce.vo.QinyiMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @description:
 * @author: sym
 * @create: 2021-11-29 22:20
 **/
@Component
@Slf4j
public class KafkaConsumer {

    private static final String TOPIC="sym-topic";
    private static final String GROUP_ID="sym-group";


    private final ObjectMapper mapper;

    public KafkaConsumer(ObjectMapper mapper) {
        this.mapper = mapper;
    }

    @KafkaListener(topics = {"qinyi-springboot"},groupId = "qinyi-springboot-kafka")
    public void listener01(ConsumerRecord<String,String> record) throws Exception{
        String key = record.key();
        String value = record.value();
        QinyiMessage qinyiMessage = mapper.readValue(value, QinyiMessage.class);
        log.info("consumer the message :value:{},groupId:qinyi-springboot-kafka",qinyiMessage.toString());
    }

    @KafkaListener(topics = {"qinyi-springboot"}, groupId = "qinyi-springboot-kafka-1")
    public void listener02(ConsumerRecord<?, ?> record) throws Exception {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()){
            Object message = kafkaMessage.get();
            //如果不一定是qinyimessage,会抛出异常，定义??类型需要try catch
            QinyiMessage qinyiMessage = mapper.readValue(message.toString(), QinyiMessage.class);
            log.info("consumer the message :value:{},groupId:qinyi-springboot-kafka-1",qinyiMessage.toString());
        }
    }
}
